package cn.itcast.demo;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
// kafka的消费者的代码
public class KafkaConsumerTest {

    public static void main(String[] args) {
        //1. 创建kafka的消费者核心类对象:  KafkaConsumer

        //1.1: 设置消费者的配置信息
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "node1:9092");
        props.setProperty("group.id", "test"); // 消费者组
        props.setProperty("enable.auto.commit", "true"); // 自动提交 偏移量信息
        props.setProperty("auto.commit.interval.ms", "1000"); // 每次提交间隔时间
        // 反序列类
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        //2. 设置监听那些topic
        consumer.subscribe(Arrays.asList("test3"));

        //3. 获取消息数据
        while (true) {
            //3.1: 通过消费者对象, 获取消息
            ConsumerRecords<String, String> records = consumer.poll(100);

            //3.2: 遍历获取每一个数据
            for (ConsumerRecord<String, String> record : records){
                System.out.println("数据为:"+record.value() + "消息偏移量:"+record.offset());
            }

        }


    }
}
